package com.wande.dataplatform.service.impl;

import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.wande.common.core.exception.ServiceException;
import com.wande.dataplatform.config.DolphinSchedulerProperties;
import com.wande.dataplatform.domain.Job;
import com.wande.dataplatform.service.IDolphinSchedulerService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * DolphinScheduler集成服务实现
 *
 * @author wande
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DolphinSchedulerServiceImpl implements IDolphinSchedulerService {

    private final DolphinSchedulerProperties properties;

    @Override
    public Long createWorkflow(Job job) {
        if (!properties.isEnabled()) {
            log.warn("DolphinScheduler未启用，跳过工作流创建");
            return null;
        }

        try {
            String url = properties.getApiUrl() + "/projects/" + properties.getProjectCode() + "/process-definitions";
            
            JSONObject requestBody = new JSONObject();
            requestBody.set("name", job.getName());
            requestBody.set("description", job.getDescription());
            requestBody.set("taskDefinitionJson", job.getDagConfig());
            requestBody.set("scheduleJson", job.getScheduleConfig());

            HttpResponse response = HttpRequest.post(url)
                .header("token", properties.getToken())
                .body(requestBody.toString())
                .execute();

            if (response.isOk()) {
                JSONObject result = JSONUtil.parseObj(response.body());
                if (result.getBool("success", false)) {
                    return result.getJSONObject("data").getLong("id");
                }
            }

            throw new ServiceException("创建DolphinScheduler工作流失败: " + response.body());
        } catch (Exception e) {
            log.error("创建DolphinScheduler工作流异常", e);
            throw new ServiceException("创建工作流失败: " + e.getMessage());
        }
    }

    @Override
    public void updateWorkflow(Job job) {
        if (!properties.isEnabled() || job.getDsWorkflowId() == null) {
            return;
        }

        try {
            String url = properties.getApiUrl() + "/projects/" + properties.getProjectCode() 
                + "/process-definitions/" + job.getDsWorkflowId();
            
            JSONObject requestBody = new JSONObject();
            requestBody.set("name", job.getName());
            requestBody.set("description", job.getDescription());
            requestBody.set("taskDefinitionJson", job.getDagConfig());
            requestBody.set("scheduleJson", job.getScheduleConfig());

            HttpResponse response = HttpRequest.put(url)
                .header("token", properties.getToken())
                .body(requestBody.toString())
                .execute();

            if (!response.isOk()) {
                throw new ServiceException("更新DolphinScheduler工作流失败: " + response.body());
            }
        } catch (Exception e) {
            log.error("更新DolphinScheduler工作流异常", e);
            throw new ServiceException("更新工作流失败: " + e.getMessage());
        }
    }

    @Override
    public void deleteWorkflow(Long workflowId) {
        if (!properties.isEnabled() || workflowId == null) {
            return;
        }

        try {
            String url = properties.getApiUrl() + "/projects/" + properties.getProjectCode() 
                + "/process-definitions/" + workflowId;

            HttpResponse response = HttpRequest.delete(url)
                .header("token", properties.getToken())
                .execute();

            if (!response.isOk()) {
                log.warn("删除DolphinScheduler工作流失败: {}", response.body());
            }
        } catch (Exception e) {
            log.error("删除DolphinScheduler工作流异常", e);
        }
    }

    @Override
    public Long startWorkflow(Long workflowId) {
        if (!properties.isEnabled() || workflowId == null) {
            throw new ServiceException("DolphinScheduler未启用或工作流ID为空");
        }

        try {
            String url = properties.getApiUrl() + "/projects/" + properties.getProjectCode() 
                + "/executors/start-process-instance";
            
            JSONObject requestBody = new JSONObject();
            requestBody.set("processDefinitionCode", workflowId);
            requestBody.set("failureStrategy", "CONTINUE");
            requestBody.set("warningType", "NONE");

            HttpResponse response = HttpRequest.post(url)
                .header("token", properties.getToken())
                .body(requestBody.toString())
                .execute();

            if (response.isOk()) {
                JSONObject result = JSONUtil.parseObj(response.body());
                if (result.getBool("success", false)) {
                    return result.getLong("data");
                }
            }

            throw new ServiceException("启动DolphinScheduler工作流失败: " + response.body());
        } catch (Exception e) {
            log.error("启动DolphinScheduler工作流异常", e);
            throw new ServiceException("启动工作流失败: " + e.getMessage());
        }
    }

    @Override
    public void stopWorkflow(Long instanceId) {
        if (!properties.isEnabled() || instanceId == null) {
            return;
        }

        try {
            String url = properties.getApiUrl() + "/projects/" + properties.getProjectCode() 
                + "/executors/execute";
            
            JSONObject requestBody = new JSONObject();
            requestBody.set("processInstanceId", instanceId);
            requestBody.set("executeType", "STOP");

            HttpResponse response = HttpRequest.post(url)
                .header("token", properties.getToken())
                .body(requestBody.toString())
                .execute();

            if (!response.isOk()) {
                throw new ServiceException("停止DolphinScheduler工作流失败: " + response.body());
            }
        } catch (Exception e) {
            log.error("停止DolphinScheduler工作流异常", e);
            throw new ServiceException("停止工作流失败: " + e.getMessage());
        }
    }

    @Override
    public String getWorkflowInstanceStatus(Long instanceId) {
        if (!properties.isEnabled() || instanceId == null) {
            return "UNKNOWN";
        }

        try {
            String url = properties.getApiUrl() + "/projects/" + properties.getProjectCode() 
                + "/process-instances/" + instanceId;

            HttpResponse response = HttpRequest.get(url)
                .header("token", properties.getToken())
                .execute();

            if (response.isOk()) {
                JSONObject result = JSONUtil.parseObj(response.body());
                if (result.getBool("success", false)) {
                    JSONObject data = result.getJSONObject("data");
                    String state = data.getStr("state");
                    return mapDolphinSchedulerStatus(state);
                }
            }

            return "UNKNOWN";
        } catch (Exception e) {
            log.error("获取DolphinScheduler工作流状态异常", e);
            return "UNKNOWN";
        }
    }

    @Override
    public String getWorkflowInstanceLogs(Long instanceId) {
        if (!properties.isEnabled() || instanceId == null) {
            return "";
        }

        try {
            String url = properties.getApiUrl() + "/log/detail";
            
            HttpResponse response = HttpRequest.get(url)
                .header("token", properties.getToken())
                .form("processInstanceId", instanceId)
                .execute();

            if (response.isOk()) {
                JSONObject result = JSONUtil.parseObj(response.body());
                if (result.getBool("success", false)) {
                    return result.getJSONObject("data").getStr("message", "");
                }
            }

            return "";
        } catch (Exception e) {
            log.error("获取DolphinScheduler工作流日志异常", e);
            return "";
        }
    }

    @Override
    public void syncInstanceStatus(Long instanceId) {
        if (!properties.isEnabled() || instanceId == null) {
            return;
        }

        try {
            String status = getWorkflowInstanceStatus(instanceId);
            log.debug("同步DolphinScheduler实例状态: instanceId={}, status={}", instanceId, status);
        } catch (Exception e) {
            log.error("同步DolphinScheduler实例状态异常", e);
        }
    }

    @Override
    public java.util.List<com.wande.dataplatform.domain.vo.JobInstanceVO> getSubTaskInstances(Long instanceId) {
        java.util.List<com.wande.dataplatform.domain.vo.JobInstanceVO> subInstances = new java.util.ArrayList<>();
        
        if (!properties.isEnabled() || instanceId == null) {
            return subInstances;
        }

        try {
            String url = properties.getApiUrl() + "/projects/" + properties.getProjectCode() 
                + "/process-instances/" + instanceId + "/tasks";

            HttpResponse response = HttpRequest.get(url)
                .header("token", properties.getToken())
                .execute();

            if (response.isOk()) {
                JSONObject result = JSONUtil.parseObj(response.body());
                if (result.getBool("success", false)) {
                    cn.hutool.json.JSONArray tasks = result.getJSONArray("data");
                    for (int i = 0; i < tasks.size(); i++) {
                        JSONObject task = tasks.getJSONObject(i);
                        com.wande.dataplatform.domain.vo.JobInstanceVO subInstance = new com.wande.dataplatform.domain.vo.JobInstanceVO();
                        subInstance.setId(task.getLong("id"));
                        subInstance.setJobName(task.getStr("name"));
                        subInstance.setStatus(mapDolphinSchedulerStatus(task.getStr("state")));
                        subInstance.setStartTime(task.getLocalDateTime("startTime", null));
                        subInstance.setEndTime(task.getLocalDateTime("endTime", null));
                        subInstances.add(subInstance);
                    }
                }
            }
        } catch (Exception e) {
            log.error("获取DolphinScheduler子任务实例异常", e);
        }

        return subInstances;
    }

    @Override
    public String getInstanceLogs(Long instanceId) {
        return getWorkflowInstanceLogs(instanceId);
    }

    /**
     * 映射DolphinScheduler状态到系统状态
     */
    private String mapDolphinSchedulerStatus(String dsStatus) {
        if (StrUtil.isEmpty(dsStatus)) {
            return "UNKNOWN";
        }

        return switch (dsStatus.toUpperCase()) {
            case "RUNNING_EXECUTION", "SUBMITTED_SUCCESS", "DISPATCH", "READY_PAUSE", "READY_STOP" -> "RUNNING";
            case "SUCCESS" -> "SUCCESS";
            case "FAILURE", "NEED_FAULT_TOLERANCE", "KILL" -> "FAILED";
            case "STOP", "PAUSE" -> "STOPPED";
            default -> "UNKNOWN";
        };
    }
}
